package com.xq.jltx.learnflow

import kotlinx.coroutines.delay
import kotlinx.coroutines.flow.*
import kotlinx.coroutines.runBlocking

/**
 * @author aidenquan
 *         Created by junlintianxia on 2021/08/11.
 */

fun main() = runBlocking<Unit> {
//    flowCreateRun()
//    flowOfCreateRun()
//    asFlowCreateRun()
    channelFlowCreateRun()
}

private suspend fun flowCreateRun() {
    flowCreate().collect {
        println("Flow collect $it")
    }
}

private suspend fun flowOfCreateRun() {
    flowOfCreate().collect {
        println("Flow collect $it")
    }
}


private suspend fun asFlowCreateRun() {
    asFlowCreate().collect {
        println("Flow collect $it")
    }
    asFlowCreate2().collect {
        println("Flow collect $it")
    }
}

private suspend fun channelFlowCreateRun() {
    channelFlowCreate().onCompletion {
        println("Flow onCompletion end")
    }.collect {
        if (it == 4) throw Exception("抛异常")
        println("Flow collect $it")
    }
}

/**
 * 第一种：flow{} builder方式创建
 */
fun flowCreate(): Flow<Int> = flow {
    println("Flow started")
    (1..5).forEach {
        delay(1000)
        emit(it)
    }
}

/**
 * 第二种:flowOf()
 */
fun flowOfCreate(): Flow<Int> = flowOf(1, 2, 3, 4, 5).onEach {
    delay(1000)
}

/**
 * 第三种asFlow()
 */
fun asFlowCreate(): Flow<Int> = (1..5).asFlow().onEach {
    delay(1000)
}

fun asFlowCreate2(): Flow<Int> = listOf(1, 2, 3, 4, 5).asFlow().onEach {
    delay(1000)
}

/**
 * 第四种channelFlow
 * flow 是 Cold Stream。在没有切换线程的情况下，生产者和消费者是同步非阻塞的。
 * channel 是 Hot Stream。而 channelFlow 实现了生产者和消费者异步非阻塞模型
 */
fun channelFlowCreate(): Flow<Int> = channelFlow<Int> {
    (1..5).forEach {
        delay(1000)
        send(it)
    }
}